package cn.ac.iie.sparkdemo;

import gtextinputformat.GTextInputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.ArrayWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn>
 * @date 2016-9-9 10:33:32
 * @version V1.0
 * @desc
 */
public class Main {

    public static void main(String[] args) {
        System.out.println("path:" + args[0]);
        Configuration configuration = new Configuration();
        configuration.set("encode", "utf-8");
        configuration.set("record_delimiter", "#");
        configuration.set("field_delimiter", ",");
        configuration.set("soround_delimiter", "");
        SparkConf conf = new SparkConf().setAppName("test");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaPairRDD<LongWritable, ArrayWritable> newAPIHadoopRDD = sc.newAPIHadoopFile(args[0],
                GTextInputFormat.class, LongWritable.class, ArrayWritable.class, configuration);
        newAPIHadoopRDD.count();
        newAPIHadoopRDD.foreach(new VoidFunction<Tuple2<LongWritable, ArrayWritable>>() {
            @Override
            public void call(Tuple2<LongWritable, ArrayWritable> t) throws Exception {
                LongWritable key = t._1;
                ArrayWritable value = t._2;
                System.out.println("data:" + key + "\t" + value);
            }
        });
    }

}
